package com.wuwangfu.func;

import com.wuwangfu.entity.ItemEventCount;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-08-12
 */
public class GoodsWindowFunc implements WindowFunction<Long, ItemEventCount, Tuple2<String,String>, TimeWindow> {
    @Override
    public void apply(Tuple2<String, String> key, TimeWindow window, Iterable<Long> input, Collector<ItemEventCount> out) throws Exception {

        long startTime = window.getStart();
        long endTime = window.getEnd();
        Long count = input.iterator().next();

        String itemId = key.f0;
        String type = key.f1;

        ItemEventCount item = ItemEventCount.of(itemId, type, startTime, endTime, count);
        out.collect(item);
    }
}
